package com.ccc.mq.dynamic.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.ccc.mq.dynamic.config.ConsumerConfig;
import com.ccc.mq.dynamic.entity.RocketInfo;
import com.ccc.mq.dynamic.entity.RocketTopicMsgModel;
import com.ccc.mq.dynamic.mapper.RocketInfoMapper;
import com.ccc.mq.dynamic.mapper.RocketTopicMsgModelMapper;
import com.ccc.mq.dynamic.pojo.RocketTopicMsgVO;
import com.ccc.mq.dynamic.service.RoketConsumerService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Objects;

/**
 * 消费者动态处理类
 *
 * @author qiaofeng
 */
@Service
@Slf4j
public class RocketRoketConsumerServiceImpl implements RoketConsumerService {

    @Autowired
    RocketInfoMapper rocketInfoMapper;

    @Autowired
    RocketTopicMsgModelMapper rocketTopicMsgModelMapper;

    @Override
    public List<RocketTopicMsgVO> list() {
        return rocketTopicMsgModelMapper.list(1);
    }

    @Override
    public void startConsumer(RocketTopicMsgVO vo) {
        Long rocketTopicId = vo.getRocketTopicId();
        if (ConsumerConfig.consumerMap.containsKey(rocketTopicId)) {
            return;
        }
        try {
            DefaultMQPushConsumer mQPushConsumer = addConsumerIntoMap(vo);
            ConsumerConfig.consumerMap.put(rocketTopicId, mQPushConsumer);
        } catch (MQClientException e) {
            log.error("创建消费者失败，原因：{}", e.getErrorMessage());
        }
    }

    @Override
    public void startConsumer(Long rocketTopicId) {
        if (ConsumerConfig.consumerMap.containsKey(rocketTopicId)) {
            return;
        }
        RocketTopicMsgModel rocketTopicMsgModel = getRocketTopicMsgModel(rocketTopicId);
        RocketInfo rocketInfo = rocketInfoMapper.selectOne(new QueryWrapper<RocketInfo>().eq("id", rocketTopicMsgModel.getRocketId()));
        try {
            DefaultMQPushConsumer mQPushConsumer = addConsumerIntoMap(rocketInfo.getAddrIp(), rocketTopicMsgModel);
            ConsumerConfig.consumerMap.put(rocketTopicId, mQPushConsumer);
        } catch (MQClientException e) {
            log.error("创建消费者失败，原因：{}", e.getErrorMessage());
        }
    }

    @Override
    public void removeConsumer(Long rocketTopicId) {
        if (ConsumerConfig.consumerMap.containsKey(rocketTopicId)) {
            ConsumerConfig.consumerMap.get(rocketTopicId).shutdown();
            ConsumerConfig.consumerMap.remove(rocketTopicId);
        }
    }

    @Override
    public void restartConsumer(Long rocketTopicId) {
        if (!ConsumerConfig.consumerMap.containsKey(rocketTopicId)) {
            return;
        }
        DefaultMQPushConsumer mqPushConsumer = ConsumerConfig.consumerMap.get(rocketTopicId);
        mqPushConsumer.shutdown();

        ConsumerConfig.consumerMap.remove(rocketTopicId);
        startConsumer(rocketTopicId);
    }

    private RocketTopicMsgModel getRocketTopicMsgModel(Long rocketTopicId) {
        RocketTopicMsgModel rocketTopicMsgModel = rocketTopicMsgModelMapper.selectOne(new QueryWrapper<RocketTopicMsgModel>().eq("id", rocketTopicId));
        if (rocketTopicMsgModel == null) {
            throw new RuntimeException("该消费者id不存在" + rocketTopicId);
        }
        return rocketTopicMsgModel;
    }

    /**
     * 创建对应的消费者服务
     */
    private DefaultMQPushConsumer addConsumerIntoMap(RocketTopicMsgVO vo) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(vo.getConsumerGroup());
        consumer.setNamesrvAddr(vo.getAddrIp());
        String subExpression = vo.getSubExpression();
        if (StringUtils.isEmpty(subExpression)) {
            subExpression = "*";
        }
        consumer.subscribe(vo.getTopicName(), subExpression);
        if (vo.getOrderly()) {
            registerOrderlyListener(consumer);
        } else {
            registerListener(consumer);
        }
        consumer.setMessageModel(vo.getMsgModelType() == 1 ? MessageModel.CLUSTERING : MessageModel.BROADCASTING);

        // 启动消费者实例
        consumer.start();
        return consumer;
    }

    /**
     * 创建对应的消费者服务
     *
     * @param addrIp              消息队列服务ip:端口
     * @param rocketTopicMsgModel 消费者需要消费的队列
     */
    private DefaultMQPushConsumer addConsumerIntoMap(String addrIp, RocketTopicMsgModel rocketTopicMsgModel) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(rocketTopicMsgModel.getConsumerGroup());
        consumer.setNamesrvAddr(addrIp);
        String subExpression = rocketTopicMsgModel.getSubExpression();
        if (StringUtils.isEmpty(subExpression)) {
            subExpression = "*";
        }
        consumer.subscribe(rocketTopicMsgModel.getTopicName(), subExpression);
        if (rocketTopicMsgModel.getOrderly()) {
            registerOrderlyListener(consumer);
        } else {
            registerListener(consumer);
        }
        consumer.setMessageModel(rocketTopicMsgModel.getMsgModelType() == 1 ? MessageModel.CLUSTERING : MessageModel.BROADCASTING);

        // 配置对应消费者组的最大消息重试次数为 10 次
        consumer.setMaxReconsumeTimes(10);

        // 启动消费者实例
        consumer.start();
        return consumer;
    }

    private void registerListener(DefaultMQPushConsumer consumer) {
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                try {
                    if (msgs.size() > 0) {
                        MessageExt messageExt = msgs.get(0);
                        consumerMsg(new String(messageExt.getBody(), StandardCharsets.UTF_8));
                    }
                } catch (Throwable e) {
                    log.error("发生异常:{}", e.getMessage());
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    }

    //消费消息
    private void consumerMsg(String msg) {
        log.info("收到的msg:{}", msg);
    }

    private void registerOrderlyListener(DefaultMQPushConsumer consumer) {
        consumer.registerMessageListener(new MessageListenerOrderly() {

            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                context.setAutoCommit(true);
                System.out.printf("%s 收到的msg: %s %n", Thread.currentThread().getName(), msgs);

                if (msgs.size() > 0) {
                    MessageExt messageExt = msgs.get(0);
                    // 消费
                    try {
                        log.info(messageExt.getTopic());
                        boolean result = consumerOrderlyMsg(new String(messageExt.getBody(), StandardCharsets.UTF_8));
                        if (Objects.equals("topic1", messageExt.getTopic())) {
                            result = true;
                        }
                        if (result) {
                            return ConsumeOrderlyStatus.SUCCESS;
                        } else {
                            context.setSuspendCurrentQueueTimeMillis(3000);
                            return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
//                            log.error("记录result = flase");
//                            return ConsumeOrderlyStatus.SUCCESS;
                        }
                    } catch (Exception e) {
//                        context.setSuspendCurrentQueueTimeMillis(3000);
//                        return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                        log.error("记录异常，", e);
                        return ConsumeOrderlyStatus.SUCCESS;
                    }
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
    }

    //消费消息
    private boolean consumerOrderlyMsg(String msg) {
        log.info("收到的msg:{}", msg);
//        throw new RuntimeException("模拟异常产生！");
        return false;
    }

}